package main

import (
	"context"
	"encoding/json"
	"fmt"
	"strconv"
	"sync"
	"time"

	sarama "github.com/Shopify/sarama"
)

type consumerGroupHandler struct {
	name   string
	client ClientHandle
}

func (consumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
func (consumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }

// 消费方式的定义
func (h *consumerGroupHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	data := make(chan *sarama.ConsumerMessage, 5)

	h.client.Init()
	i := 0
	var TotalEndTime int64
	var TotalStartTime int64

	// p := "./fmt/test.fmt"
	// logfile, err := os.OpenFile(p, os.O_RDWR|os.O_CREATE, 0666)
	// if err != nil {
	// 	fmt.Println(err)
	// }
	// defer logfile.Close()
	// fmt.SetOutput(logfile)

	for i := 0; i < 1; i++ {
		go h.DealQuintetsMsg(data, i) // 启动协程处理一个区块中所有的五元组消息
	}

	fmt.Printf("cluster [%d] -> go DealQuintetsMsg\n", *ClusterID)
	for msg := range claim.Messages() {
		fmt.Printf("cluster [%d] -> receive msg\n", *ClusterID)
		if msg.Value == nil {
			fmt.Println("nil msg")
		} else {
			data <- msg
			sess.MarkMessage(msg, "")
			sess.Commit()
			i++
			if i == 60 {
				TotalEndTime = time.Now().UnixNano()
				fmt.Println("总计用时", float64((TotalEndTime-TotalStartTime)/1e6))
			}
		}
	}
	return nil
}

// 处理一个区块中所有的五元组消息
func (h *consumerGroupHandler) DealQuintetsMsg(mmsg chan *sarama.ConsumerMessage, i int) {
	fmt.Printf("cluster [%d] -> cal task start thread %d\n", *ClusterID, i)
	for {
		msg := <-mmsg
		fmt.Printf("cluster [%d] -> thread %d start cal task at offset: %d\n", *ClusterID, i, msg.Offset)
		// time.Sleep(1 * time.Second)
		var q Quintets
		json.Unmarshal(msg.Value, &q)
		fmt.Printf("cluster [%d] -> read block [%d] 5-groups from kafka\n", *ClusterID, q.Quin[0].BlockId)
		var wg sync.WaitGroup
		startTime := time.Now()
		ObjNumCalculted := 0
		for _, quintet := range q.Quin { // 遍历本区块中所有的数据布局信息
			if (int8)(*ClusterID) == quintet.Compute { // 本机构为此纠删码条带的计算机构，起协程计算此条带的校验块
				wg.Add(1)
				ObjNumCalculted += len(quintet.Storage)
				go h.StoreRedundancyObj(&wg, quintet)
			}
		}
		wg.Wait()
		duration := time.Since(startTime).Milliseconds()
		throughput := float64(ObjectSize*ObjNumCalculted) / float64(1024*1024*duration/1000)
		fmt.Printf("cluster [%d] -> cal obj num [%d] of block [%d], cost time: %dms, throughput: %f MB/s\n", *ClusterID, ObjNumCalculted, q.Quin[0].BlockId, duration, throughput)
		BlockHeightRedundunt = q.Quin[0].BlockId // 记录最新处理的区块高度
	}
}

// 计算纠删码条带的校验块
func (h *consumerGroupHandler) StoreRedundancyObj(wg *sync.WaitGroup, quintet Quintet) {
	defer wg.Done()

	curTime := time.Now()
	fmt.Printf("cluster [%d] -> start_deal_stripe [%d] in block[%d], time: %v, timestamp: %v\n", *ClusterID, quintet.Number, quintet.BlockId, curTime.Format("2006-01-02 15:04:05.000"), curTime.UnixMilli())

	if len(quintet.Objs) == 3 { // 条带中有三个数据块
		if len(quintet.Storage) == 1 { // 来自三家不同机构 -> 3 + 1
			objData := make([][]byte, 4)
			var err error
			for i := 0; i < 3; i++ {
				// 从各个机构获取对象数据
				objData[i], err = h.client.GetObjFromClusterTLS(string(quintet.Objs[i].Oid), int(quintet.Objs[i].From))
				if err != nil {
					fmt.Printf("cluster [%d] -> get data from cluster [%d] error: %s\n", *ClusterID, quintet.Objs[i].From, err)
					return
				}
			}
			fmt.Printf("cluster [%d] -> finish get data from other clusters\n", *ClusterID)
			err = ErasureEncode(3, 1, objData) // 计算 3+1 纠删码
			if err != nil {
				fmt.Printf("cluster [%d] -> error cal erasure obj, err: %s\n", *ClusterID, err)
				return
			}
			// err = ErasureEncodeXor(3, 1, objData) // 使用Xor计算3+1纠删码
			// if err != nil {
			// 	fmt.Println(err.Error())
			// 	return
			// }
			redundancyObjId := "@@" + strconv.Itoa(int(quintet.Objs[0].From)) + "_" + string(quintet.Objs[0].Oid) + "@" + strconv.Itoa(int(quintet.Objs[1].From)) + "_" + string(quintet.Objs[1].Oid) + "@" + strconv.Itoa(int(quintet.Objs[2].From)) + "_" + string(quintet.Objs[2].Oid)
			// 存储纠删码校验块到存储方
			err = h.client.PutObjToClusterTLS(redundancyObjId, int(quintet.Storage[0]), objData[3]) // "@@"表示冗余块
			if err != nil {
				fmt.Printf("cluster [%d] -> error when write: %s(len: %d) to cluster [%d]: %s\n", *ClusterID, redundancyObjId, len(objData[3]), int(quintet.Storage[0]), err)
				return
			}
		} else if len(quintet.Storage) == 2 { // 来自两家不同机构, 其中第一个和第二个来自同一家机构 -> 两个 2 + 1
			var err error
			objData := make([][]byte, 3)
			// 第1个和第3个计算 2+1 检验块
			for k, i := range []int{0, 2} {
				objData[k], err = h.client.GetObjFromClusterTLS(string(quintet.Objs[i].Oid), int(quintet.Objs[i].From))
				if err != nil {
					fmt.Printf("cluster [%d] -> get data from cluster [%d] error: %s\n", *ClusterID, quintet.Objs[i].From, err)
					return
				}
			}
			err = ErasureEncode(2, 1, objData) // 计算 2+1 纠删码
			// err = ErasureEncodeXor(2, 1, objData) // 使用Xor计算2+1纠删码
			if err != nil {
				fmt.Println(err.Error())
				return
			}
			redundancyObjId := "@@" + strconv.Itoa(int(quintet.Objs[0].From)) + "_" + string(quintet.Objs[0].Oid) + "@" + strconv.Itoa(int(quintet.Objs[2].From)) + "_" + string(quintet.Objs[2].Oid)
			err = h.client.PutObjToClusterTLS(redundancyObjId, int(quintet.Storage[0]), objData[2]) // "@@"表示冗余块
			if err != nil {
				fmt.Printf("cluster [%d] -> error when write: %s(len: %d) to cluster [%d]: %s\n", *ClusterID, redundancyObjId, len(objData[2]), int(quintet.Storage[0]), err)
				return
			}
			// 第2个和第3个计算 2+1 检验块
			objData[0], err = h.client.GetObjFromClusterTLS(string(quintet.Objs[1].Oid), int(quintet.Objs[1].From))
			if err != nil {
				fmt.Printf("cluster [%d] -> get data from cluster [%d] error: %s, \n", *ClusterID, quintet.Objs[1].From, err)
				return
			}
			err = ErasureEncode(2, 1, objData) // 计算 2+1 纠删码
			// err = ErasureEncodeXor(2, 1, objData) // 使用Xor计算2+1纠删码
			if err != nil {
				fmt.Println(err.Error())
				return
			}
			redundancyObjId = "@@" + strconv.Itoa(int(quintet.Objs[1].From)) + "_" + string(quintet.Objs[1].Oid) + "@" + strconv.Itoa(int(quintet.Objs[2].From)) + "_" + string(quintet.Objs[2].Oid)
			err = h.client.PutObjToClusterTLS(redundancyObjId, int(quintet.Storage[1]), objData[2]) // "@@"表示冗余块
			if err != nil {
				fmt.Printf("cluster [%d] -> error when write: %s(len: %d) to cluster [%d]: %s\n", *ClusterID, redundancyObjId, len(objData[2]), int(quintet.Storage[1]), err)
				return
			}
		} else if len(quintet.Storage) == 3 { // 均来自一家机构 -> 三个 1 + 1
			for i := 0; i < 3; i++ {
				objData, err := h.client.GetObjFromClusterTLS(string(quintet.Objs[i].Oid), int(quintet.Objs[i].From))
				if err != nil {
					fmt.Printf("cluster [%d] -> get data from cluster [%d] error: %s\n", *ClusterID, quintet.Objs[i].From, err)
					continue
				}
				redundancyObjId := "@@" + strconv.Itoa(int(quintet.Objs[i].From)) + "_" + string(quintet.Objs[i].Oid)
				err = h.client.PutObjToClusterTLS(redundancyObjId, int(quintet.Storage[i]), objData) // "@@"表示冗余块
				if err != nil {
					fmt.Printf("cluster [%d] -> error when write: %s(len: %d) to cluster [%d]: %s\n", *ClusterID, redundancyObjId, len(objData), int(quintet.Storage[i]), err)
					continue
				}
			}
		} else {
			fmt.Printf("cluster [%d] -> error 5-group when write 3+1 / 2+1 redundancy obj: length of storage array error\n", *ClusterID)
			return
		}
	} else if len(quintet.Objs) == 1 { // 同一区块内剩余对象数小于3时，生成的五元组中对象数为1，使用 1 + 1 容错
		if len(quintet.Storage) == 1 {
			objData, err := h.client.GetObjFromClusterTLS(string(quintet.Objs[0].Oid), int(quintet.Objs[0].From))
			if err != nil {
				fmt.Printf("cluster [%d] -> get data from cluster [%d] error: %s\n", *ClusterID, quintet.Objs[0].From, err)
				return
			}
			redundancyObjId := "@@" + strconv.Itoa(int(quintet.Objs[0].From)) + "_" + string(quintet.Objs[0].Oid)
			err = h.client.PutObjToClusterTLS(redundancyObjId, int(quintet.Storage[0]), objData) // "@@"表示冗余块
			if err != nil {
				fmt.Printf("cluster [%d] -> error when write: %s(len: %d) to cluster [%d]: %s\n", *ClusterID, redundancyObjId, len(objData), int(quintet.Storage[0]), err)
				return
			}
		} else {
			fmt.Printf("cluster [%d] -> error 5-group when write 1+1 redundancy obj: storage array len to long\n", *ClusterID)
			return
		}
	} else {
		fmt.Printf("cluster [%d] -> error 5-group when write 1+1 redundancy obj: objs array len error\n", *ClusterID)
		return
	}
	curTime = time.Now()
	fmt.Printf("cluster [%d] -> end_deal_stripe [%d] in block[%d], time: %v, timestamp: %v\n", *ClusterID, quintet.Number, quintet.BlockId, curTime.Format("2006-01-02 15:04:05.000"), curTime.UnixMilli())
}

// 从Kafka获取数据对象五元组，并统筹计算、存储纠删码校验块到相应机构
func CalStoreParityBlock() {
	fmt.Printf("cluster [%d] start deal 5-group from kafka ...\n", *ClusterID)

	config := sarama.NewConfig()
	config.Consumer.Return.Errors = false
	config.Version = sarama.V0_11_0_2
	config.Consumer.Offsets.AutoCommit.Enable = false
	config.Consumer.Offsets.Initial = sarama.OffsetNewest

	// fmt.Println(KafkaIPPort())
	group, err := sarama.NewConsumerGroup([]string{KafkaIPPort()}, "t3", config)
	if err != nil {
		panic(err)
	}
	defer group.Close()

	for {
		handler := &consumerGroupHandler{name: "sera"}
		err := group.Consume(context.Background(), []string{"QuintetQueue"}, handler)
		if err != nil {
			fmt.Println(err.Error())
		}
	}
}
